package com.xiaoze.source;

import com.xiaoze.entity.SensorReading;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.HashMap;
import java.util.Map;
import java.util.Random;

/**
 * 自定义数据源
 *
 * @author xiaoze
 * @date 2021/12/27
 */
public class MySensorSource implements SourceFunction<SensorReading> {

    /**
     * 定义标志位，控制数据产生与停止
     */
    private boolean running = true;

    @Override
    public void run(SourceContext<SensorReading> sourceContext) throws Exception {

        // 随机数生成器
        Random random = new Random();

        Map<String, Double> sensorMap = new HashMap<>(16);

        for (int i = 0; i < 10; i++) {
            sensorMap.put("sensor_" + (i+1),60 + random.nextGaussian() * 20);
        }
        while (running) {
            for (String sensorId : sensorMap.keySet()) {
                // 在当前温度随机波动
                Double temperature = sensorMap.get(sensorId) + random.nextGaussian();
                sensorMap.put(sensorId, temperature);
                sourceContext.collect(new SensorReading(sensorId, System.currentTimeMillis(), temperature));
            }
            Thread.sleep(1000L);
        }

    }

    @Override
    public void cancel() {
        running = false;
    }

}
